package com.hmc.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;

/**
 * 配置offset消费位置auto.offset.reset = earliest（最前面） | latest（最后位  默认） | none （从该消费者组的先前消费位置开始消费）
 */
public class CustomConsumer {
    public static void main(String[] args) {
        String groupId = "test";
        String[] topics = {"first"};
        Properties properties = getProperties(groupId);
        KafkaConsumer<String, String> consumer = getConsumer(properties,topics);

        //配置读取位置
//        configOffset(consumer,200);
        //通过时间读取指定位置的数据
//        configOffsetForTime(consumer);


        consume(consumer);
    }

    private static void configOffsetForTime(KafkaConsumer<String, String> consumer) {
        //按照时间读取对应的offset
        Set<TopicPartition> assignment;
        while ((assignment = consumer.assignment()).size() == 0) {
            consumer.poll(Duration.ofSeconds(1));
        }
        HashMap<TopicPartition, Long> partitionTimestapm = new HashMap<>();
        for (TopicPartition topicPartition : assignment) {
            partitionTimestapm.put(topicPartition,System.currentTimeMillis()-1000*3600*24);
        }
        //获取时间和offset列表
        Map<TopicPartition, OffsetAndTimestamp> timestampMap = consumer.offsetsForTimes(partitionTimestapm);
        for (TopicPartition topicPartition : assignment) {
            OffsetAndTimestamp timestamp = timestampMap.get(topicPartition);
            //将时间转化为对应的offset
            consumer.seek(topicPartition,timestamp.offset());
        }
    }

    private static void configOffset(KafkaConsumer<String, String> consumer,long offset) {
        Set<TopicPartition> assignment;
        //查找特定主题下对应的分区信息
        //错误：原因：在消费者连接kafka集群还未配置分区时，直接获取分区信息为空
//        assignment = consumer.assignment();

        //保证分区分配方案已经指定完毕
        while ((assignment = consumer.assignment()).size()==0){
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofSeconds(1));
        }

        //配置消费位置
        for (TopicPartition topicPartition : assignment) {
            consumer.seek(topicPartition, offset);
        }
    }

    //消费数据
    private static void consume(KafkaConsumer<String, String> consumer) {
        while (true) {
            //拉取数据，每一秒钟一次
            ConsumerRecords<String, String> message = consumer.poll(Duration.ofSeconds(1));
            message.iterator().forEachRemaining(a->{
                System.out.println(a);
            });
        }
    }

    //创建消费者
    private static KafkaConsumer<String, String> getConsumer(Properties properties, String[] topics) {
        KafkaConsumer<String,String> consumer = new KafkaConsumer(properties);
        //订阅topic
        List<String> list = Arrays.asList(topics);
        consumer.subscribe(list);
        return consumer;
    }

    //创建配置类
    private static Properties getProperties(String groupId) {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092,hadoop103:9092");
        //反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        //配置组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,groupId);
        return properties;
    }
}
